package io.kiki.sba.registry.server.clientInterface.push;


import com.google.common.collect.Sets;
import io.kiki.sba.registry.common.model.SubscriberUtils;
import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.constants.ValueConstants;
import io.kiki.sba.registry.common.model.store.*;
import io.kiki.sba.registry.core.model.ReceivedConfigData;
import io.kiki.sba.registry.core.model.Scope;
import io.kiki.sba.registry.server.clientInterface.bootstrap.Config;
import io.kiki.sba.registry.server.clientInterface.cache.CacheService;
import io.kiki.sba.registry.server.clientInterface.cache.DatumKey;
import io.kiki.sba.registry.server.clientInterface.cache.Key;
import io.kiki.sba.registry.server.clientInterface.cache.Value;
import io.kiki.sba.registry.server.clientInterface.circuit.breaker.CircuitBreakerService;
import io.kiki.sba.registry.server.clientInterface.metadata.MetadataCacheRegistry;
import io.kiki.sba.registry.server.clientInterface.multi.cluster.DataCenterCache;
import io.kiki.sba.registry.server.clientInterface.store.Subscribers;
import io.kiki.sba.registry.server.shared.util.DatumUtils;
import io.kiki.sba.registry.task.FastRejectedExecutionException;
import io.kiki.sba.registry.task.KeyedThreadPoolExecutor;
import io.kiki.sba.registry.util.DatumVersionUtil;
import io.kiki.sba.registry.util.StringUtil;
import lombok.Getter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.Map.Entry;

import static io.kiki.sba.registry.server.clientInterface.push.PushMetrics.Fetch.*;

public class FirePushService {
    private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PushProcessor.class);
    final Config config;
    final ChangeHandler changeHandler = new ChangeHandler();
    private final Set<String> dataCenterSet;
    @Autowired
    PushSwitchService pushSwitchService;
    @Resource
    CacheService sessionDatumCacheService;
    @Autowired
    Subscribers subscribers;
    @Autowired
    CircuitBreakerService circuitBreakerService;
    @Autowired
    MetadataCacheRegistry metadataCacheRegistry;
    @Autowired
    PushProcessor pushProcessor;
    @Autowired
    ChangeProcessor changeProcessor;
    @Autowired
    WatchProcessor watchProcessor;
    @Autowired
    DataCenterCache dataCenterCache;
    @Getter
    RegProcessor regProcessor;
    private KeyedThreadPoolExecutor watchPushExecutor;

    public FirePushService(Config config) {
        this.config = config;
        this.dataCenterSet = Collections.singleton(config.getDataCenter());
    }

    static void handleFireOnWatchException(Watcher watcher, Throwable e) {
        if (e instanceof FastRejectedExecutionException) {
            logger.error("failed to fireOnWatch {}, {}", watcher.shortDesc(), e.getMessage());
            return;
        }
        logger.error("failed to fireOnWatch {}", watcher.shortDesc(), e);
    }

    @PostConstruct
    public void init() {
        watchPushExecutor = new KeyedThreadPoolExecutor("WatchPushExecutor", config.getWatchPushTaskWorkerSize(), config.getWatchPushTaskMaxBufferSize());
        this.regProcessor = new RegProcessor(config.getSubscriberRegisterTaskWorkerSize(), new RegHandlerImpl(this));
    }

    public boolean fireOnChange(String dataInfoId, TriggerPushContext triggerPushContext) {
        try {
            changeProcessor.fireChange(dataInfoId, changeHandler, triggerPushContext);
            CHANGE_TASK_COUNTER.inc();
            return true;
        } catch (Throwable e) {
            logger.error("failed to fireOnChange {}, changeCtx={}", dataInfoId, triggerPushContext, e);
            return false;
        }
    }

    public boolean fireOnPushEmpty(Subscriber subscriber, String dataCenter) {
        long version = DatumVersionUtil.nextId();
        if (DatumVersionUtil.useConfregVersionGen()) {
            version = DatumVersionUtil.confregNextId(0);
        }
        return fireOnPushEmpty(subscriber, dataCenter, version);
    }

    public boolean fireOnPushEmpty(Subscriber subscriber, String dataCenter, long version) {
        long pushVersion = subscriber.markPushEmpty(dataCenter, version);

        SubDatum emptyDatum = DatumUtils.newEmptySubDatum(dataCenter, subscriber, pushVersion);
        final long now = System.currentTimeMillis();
        TriggerPushContext triggerPushContext = new TriggerPushContext(dataCenter, emptyDatum.getVersion(), null, now);
        Cause cause = new Cause(triggerPushContext, Type.empty, Collections.singletonMap(dataCenter, now));
        processPush(cause, DataCenterToSubDatumMap.of(emptyDatum), Collections.singletonList(subscriber));
        PUSH_EMPTY_COUNTER.inc();
        return true;
    }

    public boolean fireOnRegister(Subscriber subscriber) {
        try {
            REGISTER_TASK_COUNTER.inc();
            return regProcessor.fireOnReg(subscriber);
        } catch (Throwable e) {
            logger.error("failed to fireOnRegister {}", subscriber.shortDesc(), e);
            return false;
        }
    }

    public boolean fireOnWatcher(Watcher watcher, ReceivedConfigData receivedConfigData) {
        try {
            if (!pushSwitchService.canIpPushLocal(watcher.getClientUrl().getIp())) {
                return false;
            }
            int level = config.getClientNodePushConcurrencyLevel();
            Tuple tuple = Tuple.of(watcher.getClientUrl().buildAddressString(), watcher.getDataInfoId().hashCode() % level);
            watchPushExecutor.execute(tuple, new WatchTask(watcher, receivedConfigData));
            WATCH_TASK_COUNTER.inc();
            return true;
        } catch (Throwable e) {
            handleFireOnWatchException(watcher, e);
            return false;
        }
    }

    public boolean fireOnDatum(SubDatum subDatum, String dataNode) {
        try {
            DataInfo dataInfo = DataInfo.valueOf(subDatum.getDataInfoId());
            List<Subscriber> subscriberList = this.subscribers.getSubscriberList(dataInfo.getDataInfoId());
            final long now = System.currentTimeMillis();
            TriggerPushContext triggerPushContext = new TriggerPushContext(subDatum.getDataCenter(), subDatum.getVersion(), dataNode, now);
            final long datumTimestamp = Trace.getTriggerPushTimestamp(subDatum);
            final Cause cause = new Cause(triggerPushContext, Type.temp, Collections.singletonMap(subDatum.getDataCenter(), datumTimestamp));
            processPush(cause, DataCenterToSubDatumMap.of(subDatum), subscriberList);
            PUSH_TEMP_COUNTER.inc();
            return true;
        } catch (Throwable e) {
            logger.error("failed to fireOnDatum {}", subDatum, e);
            return false;
        }
    }

    boolean doExecuteOnChange(String changeDataInfoId, TriggerPushContext triggerPushContext) {
        final Map<String, Long> expectDataCenterToDatumVersionMap = triggerPushContext.getExpectDatumVersion();
        final DataCenterToSubDatumMap dataCenterToSubDatumMap = getDataCenterToSubDatumMap(changeDataInfoId, expectDataCenterToDatumVersionMap);
        if (dataCenterToSubDatumMap == null) {
            // datum change, but get null datum, should not happen
            logger.error("[changeNil] {},{}", changeDataInfoId, expectDataCenterToDatumVersionMap);
            return false;
        }
        for (Entry<String, Long> entry : expectDataCenterToDatumVersionMap.entrySet()) {
            SubDatum subDatum = dataCenterToSubDatumMap.getSubDatum(entry.getKey());
            if (subDatum == null) {
                // datum change, but get null datum, should not happen
                logger.error("[changeNil] {},{},{}", entry.getKey(), changeDataInfoId, entry.getValue());
                return false;
            }
            if (subDatum.getVersion() < entry.getValue()) {
                logger.error("[changeLessVer] {},{},{}<{}", entry.getKey(), changeDataInfoId, dataCenterToSubDatumMap.getDataCenterToSubDatumVersionMap(), entry.getValue());
                return false;
            }
        }

        onChange(triggerPushContext, dataCenterToSubDatumMap);
        return true;
    }

    private void onChange(TriggerPushContext triggerPushContext, DataCenterToSubDatumMap dataCenterToSubDatumMap) {
        Map<Scope, List<Subscriber>> scopeToSubscriberListMap = SubscriberUtils.groupByScope(subscribers.getDataList(dataCenterToSubDatumMap.getDataInfoId()));

        final Map<String, Long> dataCenterToDatumTimestamp = new HashMap<>(dataCenterToSubDatumMap.getDataCenterToSubDatumMap().size());
        for (Entry<String, SubDatum> entry : dataCenterToSubDatumMap.getDataCenterToSubDatumMap().entrySet()) {
            dataCenterToDatumTimestamp.put(entry.getKey(), Trace.getTriggerPushTimestamp(entry.getValue()));
        }
        final Cause cause = new Cause(triggerPushContext, Type.sub, dataCenterToDatumTimestamp);
        for (Map.Entry<Scope, List<Subscriber>> entry : scopeToSubscriberListMap.entrySet()) {
            processPush(cause, dataCenterToSubDatumMap, entry.getValue());
        }
    }

    private boolean processPush(Cause cause, DataCenterToSubDatumMap dataCenterToSubDatumMap, List<Subscriber> subscriberList) {
        if (!pushSwitchService.canLocalDataCenterPush() || !pushSwitchService.canPushMultiDataCenter(dataCenterToSubDatumMap.dataCenterSet())) {
            return false;
        }
        if (subscriberList.isEmpty()) {
            return false;
        }
        // if pushEmpty, do not check the version
        if (cause.type != Type.empty) {
            subscriberList = subscribersPushCheck(cause, dataCenterToSubDatumMap.getDataCenterToSubDatumMap(), subscriberList);
            if (CollectionUtils.isEmpty(subscriberList)) {
                return false;
            }
        }
        Map<InetSocketAddress, Map<String, Subscriber>> inetSocketAddressToRegisterIdToSubscriberMapMap = SubscriberUtils.groupBySourceAddress(subscriberList);
        for (Map.Entry<InetSocketAddress, Map<String, Subscriber>> entry : inetSocketAddressToRegisterIdToSubscriberMapMap.entrySet()) {
            final InetSocketAddress inetSocketAddress = entry.getKey();
            final Map<String, Subscriber> registerIdToSubscriberMap = entry.getValue();
            pushProcessor.firePush(cause, inetSocketAddress, registerIdToSubscriberMap, dataCenterToSubDatumMap);
        }
        return true;
    }

    DataCenterToSubDatumMap getDataCenterToSubDatumMap(String dataInfoId, Map<String, Long> expectVersions) {
        StringUtil.checkNotEmpty(expectVersions, "expectVersions");
        Key key = new Key(DatumKey.class.getName(), new DatumKey(dataInfoId, expectVersions.keySet()));
        Value value = sessionDatumCacheService.getIfPresent(key);
        if (value == null) {
            return miss(key);
        }
        DataCenterToSubDatumMap dataCenterToSubDatumMap = (DataCenterToSubDatumMap) value.getPayload();

        if (dataCenterToSubDatumMap == null || !expectVersions.keySet().equals(dataCenterToSubDatumMap.dataCenterSet())) {
            return miss(key);
        }
        for (Entry<String, Long> entry : expectVersions.entrySet()) {
            SubDatum subDatum = dataCenterToSubDatumMap.getSubDatum(entry.getKey());
            if (subDatum != null && subDatum.getVersion() < entry.getValue()) {
                return miss(key);
            }
        }
        // the expect version got
        CACHE_HIT_COUNTER.inc();
        return dataCenterToSubDatumMap;
    }

    private DataCenterToSubDatumMap miss(Key key) {
        Value value;
        CACHE_MISS_COUNTER.inc();
        // the cache is too old
        sessionDatumCacheService.invalidate(key);
        value = sessionDatumCacheService.get(key);
        return value == null ? null : (DataCenterToSubDatumMap) value.getPayload();
    }

    private List<Subscriber> subscribersPushCheck(Cause cause, Map<String, SubDatum> dataCenterToSubDatumMap, List<Subscriber> subscriberList) {
        List<Subscriber> subscribersSend = new ArrayList<>();
        for (Subscriber subscriber : subscriberList) {
            CircuitBreakerStatistic subscriberStatistic = subscriber.getStatistic();
            if (circuitBreakerService.pushCircuitBreaker(subscriberStatistic, subscriber.hasPushed())) {
                logger.info("[CircuitBreaker]subscriber:{} push check circuit break, statistic:{}", subscriber.shortDesc(), subscriberStatistic);
                continue;
            }

            Map<String, Long> dataCenterToVersionMap = new HashMap<>(dataCenterToSubDatumMap.size());
            for (Entry<String, SubDatum> entry : dataCenterToSubDatumMap.entrySet()) {
                dataCenterToVersionMap.put(entry.getKey(), entry.getValue().getVersion());
            }

            if (subscriber.isAcceptMulti()) {
                // when sub is acceptMulti and has not push,
                // only PushType.Reg can trigger push multi datacenter together
                if (!subscriber.hasPushed() && cause.type != Type.reg) {
                    continue;
                }

                // sub accept multi, check multi datacenter version,
                // return true if one of any dataCenter need to update
                if (subscriber.checkVersion(dataCenterToVersionMap)) {
                    subscribersSend.add(subscriber);
                }
            } else {
                // sub only accept local datacenter
                if (dataCenterSet.equals(dataCenterToVersionMap.keySet()) && subscriber.checkVersion(dataCenterToVersionMap)) {
                    subscribersSend.add(subscriber);
                }
            }
        }
        return subscribersSend;
    }

    boolean doExecuteOnReg(String dataInfoId, List<Subscriber> subscriberList) {

        Map<Boolean, List<Subscriber>> acceptMultiMap = SubscriberUtils.groupByMulti(subscriberList);

        Set<String> syncEnableDataCenters = metadataCacheRegistry.getPushEnableDataCenters();
        Set<String> getDatumDataCenters = new HashSet<>(syncEnableDataCenters.size() + 1);
        getDatumDataCenters.add(config.getDataCenter());
        for (String syncDataCenter : dataCenterCache.getSyncDataCenters()) {
            if (syncEnableDataCenters.contains(syncDataCenter)) {
                getDatumDataCenters.add(syncDataCenter);
            }
        }

        // accept multi sub register
        doExecuteOnReg(dataInfoId, acceptMultiMap.get(true), getDatumDataCenters);

        // not accept multi sub register
        doExecuteOnReg(dataInfoId, acceptMultiMap.get(false), Collections.singleton(config.getDataCenter()));
        return true;
    }

    private void doExecuteOnReg(String dataInfoId, List<Subscriber> subscriberList, Set<String> dataCenterSet) {
        if (CollectionUtils.isEmpty(subscriberList)) {
            return;
        }
        Map<String, Long> getDatumVersions = new HashMap<>(dataCenterSet.size());
        Map<String, Long> expectDatumVersions = new HashMap<>(dataCenterSet.size());
        Map<String, Long> datumTimestamp = new HashMap<>(dataCenterSet.size());
        for (String dataCenter : dataCenterSet) {
            getDatumVersions.put(dataCenter, Long.MIN_VALUE);
            expectDatumVersions.put(dataCenter, 0L);
            datumTimestamp.put(dataCenter, System.currentTimeMillis());
        }

        DataCenterToSubDatumMap dataCenterToSubDatumMap = getDataCenterToSubDatumMap(dataInfoId, getDatumVersions);
        if (dataCenterToSubDatumMap == null || CollectionUtils.isEmpty(dataCenterToSubDatumMap.getDataCenterToSubDatumMap())) {
            Subscriber first = subscriberList.get(0);
            dataCenterToSubDatumMap = DatumUtils.newEmptyMultiSubDatum(first, dataCenterSet, ValueConstants.DEFAULT_NO_DATUM_VERSION);
            logger.warn("[registerEmptyPush]{},{},subNum={},{}", dataInfoId, dataCenterSet, subscriberList.size(), first.shortDesc());
        } else
            if (!getDatumVersions.keySet().equals(dataCenterToSubDatumMap.dataCenterSet())) {
                Set<String> tobeAdd = Sets.difference(getDatumVersions.keySet(), dataCenterToSubDatumMap.dataCenterSet());
                Subscriber first = subscriberList.get(0);
                for (String dataCenter : tobeAdd) {
                    logger.warn("[registerLakeDatumPush]{},{},subNum={},{}", dataInfoId, dataCenterSet, subscriberList.size(), first.shortDesc());
                    dataCenterToSubDatumMap.putDatum(dataCenter, DatumUtils.newEmptySubDatum(dataCenter, first, ValueConstants.DEFAULT_NO_DATUM_VERSION));
                }
            }
        TriggerPushContext pushTriggerPushContext = new TriggerPushContext(expectDatumVersions, null, SubscriberUtils.getMinRegisterTimestamp(subscriberList));
        Cause cause = new Cause(pushTriggerPushContext, Type.reg, datumTimestamp);
        Map<Scope, List<Subscriber>> scopes = SubscriberUtils.groupByScope(subscriberList);
        for (List<Subscriber> scopeList : scopes.values()) {
            processPush(cause, dataCenterToSubDatumMap, scopeList);
        }
    }

    final class ChangeHandler implements ChangeProcessor.ChangeHandler {

        @Override
        public boolean onChange(String dataInfoId, TriggerPushContext triggerPushContext) {
            try {
                CHANGE_TASK_EXEC_COUNTER.inc();
                doExecuteOnChange(dataInfoId, triggerPushContext);
                return true;
            } catch (Throwable e) {
                logger.error("failed to do change Task, {}, {}", dataInfoId, triggerPushContext, e);
                return false;
            }
        }
    }

    final class WatchTask implements Runnable {
        final long createTimestamp = System.currentTimeMillis();
        final Watcher watcher;
        final ReceivedConfigData receivedConfigData;

        WatchTask(Watcher watcher, ReceivedConfigData receivedConfigData) {
            this.watcher = watcher;
            this.receivedConfigData = receivedConfigData;
        }

        @Override
        public void run() {
            try {
                watchProcessor.doExecuteOnWatch(watcher, receivedConfigData, createTimestamp);
            } catch (Throwable e) {
                logger.error("failed to do watch Task, {}, version={}", watcher.shortDesc(), receivedConfigData.getVersion(), e);
            }
        }
    }
}
